package com.shanks.framework.mq.rocketmq.config;

import com.shanks.framework.mq.common.util.MQUtil;
import com.shanks.framework.mq.enums.MQZone;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.rocketmq.producer.RocketMQProducerAttribute;
import com.shanks.framework.mq.rocketmq.producer.RocketMQProducerImpl;
import com.shanks.framework.mq.service.ProducerService;
import jakarta.annotation.Resource;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class RocketMQProducerInitializer {

    private static final Logger logger = LoggerFactory.getLogger(RocketMQProducerInitializer.class);

    @Resource
    private Environment environment;

    @Resource
    private ProducerService producerService;

    public RocketMQProducerImpl createAndResiterMQProducer(String propertyPrefix, MQZone mqZone) throws MQException {
        try {
            // 初始化生产者属性
            RocketMQProducerAttribute rocketMQProducerAttribute = buildProduceBean(propertyPrefix, mqZone);
            //创建并启动生产者
            DefaultMQProducer mqProducer = createAndStartMQProducer(rocketMQProducerAttribute);
            // 将生产者放入map中管理
            RocketMQProducerImpl rocketMQProducerImpl = new RocketMQProducerImpl(mqProducer, rocketMQProducerAttribute);
            rocketMQProducerImpl.setProducerKey(rocketMQProducerAttribute.getProducerKey());
            producerService.resgisterProducer(rocketMQProducerImpl.getProducerKey(), rocketMQProducerImpl);
            return rocketMQProducerImpl;
        } catch (Exception e) {
            logger.error("create RocketMQProducerImpl failed", e);
            throw new MQException(e);
        }
    }

    private RocketMQProducerAttribute buildProduceBean(String propertyPrefix, MQZone mqZone) throws MQException {
        RocketMQProducerAttribute producerBean = new RocketMQProducerAttribute();

        // 设置基础属性
        producerBean.setNamesrvAddr(getProperty(propertyPrefix + ".namesrvAddr"));
        producerBean.setCanaryNamesrvAddr(getProperty(propertyPrefix + ".canaryNamesrvAddr"));

        // 设置实例名称，支持默认值
        producerBean.setInstanceName(
                getPropertyWithDefault(propertyPrefix + ".instanceName", "rocketmq.default-producer.instanceName")
        );

        producerBean.setGroupName(getProperty(propertyPrefix + ".groupName"));

        // 设置最大消息大小，支持默认值
        producerBean.setMaxMessageSize(
                getPropertyWithDefaultAsInt(propertyPrefix + ".maxMessageSize", "rocketmq.default-producer.maxMessageSize", 0)
        );

        // 设置消息超时时间，支持默认值
        producerBean.setSendMsgTimeout(
                getPropertyWithDefaultAsInt(propertyPrefix + ".sendMsgTimeout", "rocketmq.default-producer.sendMsgTimeout", 0)
        );

        producerBean.setProducerKey(getProperty(propertyPrefix + ".producerKey"));

        // 可选的 Ack 消息主题
        Optional.ofNullable(getProperty(propertyPrefix + ".ackMsgTopic"))
                .filter(StringUtils::isNotBlank)
                .ifPresent(producerBean::setAckMsgTopic);

        // 灰度环境的特殊处理
        if (mqZone == MQZone.CANARY) {
            producerBean.setNamesrvAddr(getProperty("rocketmq.canaryNamesrvAddr"));
        }

        return producerBean;
    }

    /**
     * 从生产者属性中填充属性初始化producer
     *
     * @param rocketMQProducerAttribute
     * @return
     */
    private DefaultMQProducer createAndStartMQProducer(RocketMQProducerAttribute rocketMQProducerAttribute) throws MQException {
        DefaultMQProducer producer = new DefaultMQProducer(rocketMQProducerAttribute.getGroupName());
        producer.setNamesrvAddr(rocketMQProducerAttribute.getNamesrvAddr());
        producer.setInstanceName(rocketMQProducerAttribute.getInstanceName());
        producer.setMaxMessageSize(rocketMQProducerAttribute.getMaxMessageSize());
        producer.setSendMsgTimeout(rocketMQProducerAttribute.getSendMsgTimeout());
        try {
            producer.start();
            logger.info("RocketMQ producer started successfully. groupName:{}, namesrvAddr:{}", rocketMQProducerAttribute.getGroupName(), rocketMQProducerAttribute.getNamesrvAddr());
            String producerKey = rocketMQProducerAttribute.getProducerKey();
            MQUtil.mQProducerKeys.add(producerKey);
            logger.info("mq producer keys:{}", MQUtil.mQProducerKeys);
        } catch (MQClientException e) {
            logger.error("RocketMQ producer started failed. groupName:{}, namesrvAddr:{}", rocketMQProducerAttribute.getGroupName(), rocketMQProducerAttribute.getNamesrvAddr(), e);
            throw new MQException(e);
        }
        return producer;
    }


    /**
     * 获取属性值，如果为空则返回默认值的配置。
     */
    private String getPropertyWithDefault(String key, String defaultKey) {
        return Optional.ofNullable(getProperty(key))
                .filter(StringUtils::isNotBlank)
                .orElseGet(() -> getProperty(defaultKey));
    }

    /**
     * 获取整数属性值，如果为空则返回默认值的配置。
     */
    private int getPropertyWithDefaultAsInt(String key, String defaultKey, int defaultValue) {
        String value = getPropertyWithDefault(key, defaultKey);
        return StringUtils.isBlank(value) ? defaultValue : Integer.parseInt(value);
    }

    /**
     * 简单获取属性
     */
    private String getProperty(String key) {
        return environment.getProperty(key);
    }

}
